package com.lagou.mall.esflow.server.consumer;

import static com.lagou.mall.esflow.server.constant.ESFlowConstant.ThreadConfiguration.CORE_NUM;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.lagou.mall.esflow.client.constant.KafkaConstant;
import com.lagou.mall.esflow.server.ESFlowSystemException;
import com.lagou.mall.esflow.server.config.KafkarConfig;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class ESBinlogConsumer extends   IIndexConsumer{


    @Resource
    private KafkaTemplate kafkaTemplate;
    @Resource
    private KafkarConfig kafkarConfig;




    @PostConstruct
    void init() {
        ThreadFactory threadFactory = new ThreadFactoryBuilder()
                .setNameFormat(this.getClass().getSimpleName() + "-thread-pool %s").build();

        ThreadPoolExecutor consumerThreadPool = new ThreadPoolExecutor(
                CORE_NUM,
                CORE_NUM,
                1000,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(100), threadFactory,
                new ThreadPoolExecutor.CallerRunsPolicy());

        super.setThreadPoolExecutor(consumerThreadPool);
        super.setKafkaConsumer(kafkarConfig.createConsumer(kafkaTemplate, "item-demo"));

    }



    @KafkaListener(topics = KafkaConstant.TOPIC, groupId = "kafka-springboot-001")
    public void consumer(ConsumerRecord<String, String> record, Acknowledgment ack) throws InterruptedException {
        System.out.println(
                LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")) + "接收到kafka消息,partition:" + record.partition() + ",offset:" + record.offset() + "value:" + record.value());
        TimeUnit.SECONDS.sleep(1);
        ack.acknowledge();
    }


    @Override
    public void sleep() {
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            throw ESFlowSystemException.error("kafkaConsumer sleep  failed", e);
        }
    }
}